package com.gome.ss.flowcrawl.extractor;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Random;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.script.Invocable;
import javax.script.ScriptEngine;
import javax.script.ScriptException;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;
import org.json.JSONObject;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Attributes;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

import com.gome.ss.common.AppConfiguer;
import com.gome.ss.common.Utils;
import com.gome.ss.core.schedule.ThreadTaskBase;
import com.gome.ss.core.schedule.ThreadTaskBuilder;
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.util.JSON;

public class FlowCrawlExtractor {
	private static Logger logger = Logger.getLogger(FlowCrawlExtractor.class);
	
	private FlowCrawlExtractorConfig config = null;
	private FlowCrawlExtractorStatus status = null;
	private FlowCrawlExtractorTask task = null;
	private ScriptEngine jsEngine = null;
	
	private ConnectionFactory connectionFactory = null;
	private Connection connection = null;
	private Session session = null;
	private MessageConsumer consumer = null;
	private MessageProducer producer = null;
	private String dataMQName = "";
	
	private FlowCrawlExtractorDataParser parser = null;
	
	private HTMLFilter htmlFilter = null;
	
	private List<String> hasExceScriptIdList = new ArrayList<String>(); 
	
	public FlowCrawlExtractor() {
		this.status = new FlowCrawlExtractorStatus();
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                AppConfiguer.getStringByKey("activeMQHost"));
        
        try {
			connection = connectionFactory.createConnection();
			connection.start();
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		} catch (Exception e) {
			logger.error(Utils.getTrace(e));
		}
	}
	
	private FlowCrawlExtractor(FlowCrawlExtractorConfig config) {
		this.status = new FlowCrawlExtractorStatus(config);
		setConfig(config);
	}
	
	static public FlowCrawlExtractor newRandomExtractor(FlowCrawlExtractorConfig config) {
		return new FlowCrawlExtractor(config);
	}
	
	public void setConfig(FlowCrawlExtractorConfig config) {
		this.config = config;
		this.status.setConfig(config);
		
		if (config.isDataPushToMQ()) {
			String mqname = config.getDataMQName();
			try {
				if (producer != null && !mqname.equals(dataMQName)) {
					producer.close();
					dataMQName = mqname;
				}
				Destination destination = session.createQueue(mqname);
				producer = session.createProducer(destination);
			} catch (JMSException e) {
				logger.error(Utils.getTrace(e));
			}
		}
		
        htmlFilter = new HTMLFilter(FlowCrawlExtractorManager.getFilterDicFilePathName(config.getName()));
        loadFilterDict();
	}
	
	public void pushToDataMq(Message data) throws JMSException {
		if (producer == null)
			return;
        producer.send(data);
	}
	
	public void loadFilterDict() {
        try {
			htmlFilter.loadDict();
		} catch (Exception e) {
			logger.error(Utils.getTrace(e));
		}
	}
	
	public void setTask(FlowCrawlExtractorTask task) {
		this.task = task;
	}
	
	public FlowCrawlExtractorTask getTask() {
		return this.task;
	}
	
	public void startParser() {
		if (parser == null)
	        parser = new FlowCrawlExtractorDataParser(this);
		
		try {
			Destination destination = session.createQueue(FlowCrawlExtractorManager.mqprefix + config.getName() + "_" + config.getStepID());
			consumer = session.createConsumer(destination);
			consumer.setMessageListener(parser);
		} catch (JMSException e) {
			logger.error(Utils.getTrace(e));
		}
	}
	
	public FlowCrawlExtractorStatus getStatus() {
		return status;
	}
	
	public FlowCrawlExtractorConfig getConfig() {
		return config;
	}
	
	public boolean start() {
		if (status.getStatus() == FlowCrawlExtractorStatus.STATUS_START)
			return false;
		
		if (task == null)
			return false;
		
		DBCollection collection = AppConfiguer.flowCrawlConfigMongo.getDB(AppConfiguer.FLOWCRAWL_DB_NAME)
				 .getCollection(config.getName());
		
		BasicDBObject query = new BasicDBObject();
		query.append("stepID", config.getStepID());
		DBCursor cursor = collection.find(query);
		
		logger.info(config.getName() + " need extract pages count : " + cursor.size());
		
		status.setDateSize(cursor.count());
		status.start();
		
		ThreadTaskBase threadTask = (ThreadTaskBase) task;
		Thread thread = ThreadTaskBuilder.newThreadTask(threadTask);
		thread.start();
		
		return true;
	}
	
	public void close() {
		try {
			if (producer != null)
				producer.close();
			consumer.close();
			session.close();
			connection.close();
		} catch (JMSException e) {
			logger.error(Utils.getTrace(e));
		}
	}
	
	private class FlowCrawlExtractorDataParser implements MessageListener {
		private FlowCrawlExtractor extractor = null;
		private DBCollection collection = null;
		
		public FlowCrawlExtractorDataParser(FlowCrawlExtractor extractor) {
			this.extractor = extractor;
			this.collection = AppConfiguer.flowCrawlConfigMongo.getDB(AppConfiguer.FLOWCRAWL_DB_NAME)
					 .getCollection(extractor.getConfig().getName()+FlowCrawlExtractorManager.collectionNameSuffix);
		}

		@Override
		public void onMessage(Message message) {
			try {
				TextMessage textMess = (TextMessage)message;
				extractor.pushToDataMq(textMess);
				
				DBObject obj = (BasicDBObject) JSON.parse(textMess.getText());
				DBObject dataObj = parseData(obj);
				
				if (dataObj != null) {
					String id = (String) dataObj.get("id");
					String stepID = (String) dataObj.get("stepID");
					BasicDBObject query = new BasicDBObject();
					query.append("id", id);
					query.append("stepID", stepID);
					collection.update(query, dataObj, true, false);
				}
			} catch (JMSException e) {
				logger.error(Utils.getTrace(e));
			}
		}
	}
	
	public BasicDBObject getRandomOne() {
		
		FlowCrawlExtractorConfig config = this.getConfig();
		
		DBCollection collection = AppConfiguer.flowCrawlConfigMongo.getDB(AppConfiguer.FLOWCRAWL_DB_NAME)
				 .getCollection(this.getConfig().getName());
		
		BasicDBObject query = new BasicDBObject();
		query.append("stepID", config.getStepID());
		DBCursor cursor = collection.find(query);
		int c = cursor.count();
		
		if (c <= 0)
			return null;
		
		if (c > 1) {
			
			Random random = new Random();
			if (c < 1000)
				c = random.nextInt(c);
			else
				c = random.nextInt(1000);
			
			logger.info(config.getName() + " extract data at No." + c);
			
			cursor.skip(c-1);
		}
		
		DBObject obj = cursor.next();
		BasicDBObject dataObj = parseData(obj);
		
		if (dataObj == null) {
			logger.error(config.getName() + " extract result is null!");
			return new BasicDBObject();
		}
		
		logger.info(config.getName() + " extract test success!");
		return dataObj;
	}
	
	public BasicDBObject parseData(DBObject obj) {
		BasicDBObject dataObj = new BasicDBObject();
		dataObj.append("id", obj.get("id"));
		dataObj.append("stepID", obj.get("stepID"));
		dataObj.append("date", obj.get("date"));
		dataObj.append("url", obj.get("url"));
		dataObj.append("title", obj.get("title"));
		dataObj.append("parentPage", obj.get("parent"));
		dataObj.append("pageCookies", obj.get("cookies"));
		dataObj.append("crawlName", obj.get("crawlName"));
		
		BasicDBList facetKeys = new BasicDBList();
		for (String ikey : config.getFacetKeys()) {
			facetKeys.add(ikey);
		}
		
		if (obj.get("keyWord") != null) {
			BasicDBList keyWords = new BasicDBList();
			String keyWord = (String) obj.get("keyWord");
			if (!keyWord.equals("")) {
				String[] kwset = keyWord.split(";");
				for (String kw : kwset)
					keyWords.add(kw);
				dataObj.append("keyWord", keyWords);
				facetKeys.add("keyWord");
			}
		}
		
		facetKeys.add("crawlName");
		
		dataObj.append(FlowCrawlExtractorManager.extractDataFacetKey, facetKeys);
		
		try {
			String pageSource = (String)obj.get("pageSource");
			/*
			 *  根据重复次数清理页面中的元素
			 *  不靠谱,暂时注释掉
				pageSource = htmlFilter.filt(pageSource);
			 */
			Document doc = Jsoup.parse(pageSource);
			Collection<FlowCrawlExtractorConfigField> fileds = config.getExtractFileds().values();
			for (FlowCrawlExtractorConfigField field : fileds) {
				if (field.getType().equals("script"))
					dataObj.append(field.getKey(), getWithScript(pageSource, field.getScript(), field.getId()));
				else {
					Elements nodes = doc.select(field.getSelectQuery());
					if (nodes.size() <= 0) {
						dataObj.append(field.getKey(), "");
						logger.error("Field " + field.getKey() + " " + field.getSelectQuery() + " is empty in " +
								(String)obj.get("crawlName") + " " + (String)obj.get("stepID") + " " + (String)obj.get("id"));
						return null;
					}
					
					if (field.isMultiple()) {
						BasicDBList dataObjs = new BasicDBList();
						for (int i = 0; i < nodes.size(); i ++) {
							if (field.getType().equals("text"))
								dataObjs.add(getText(nodes.get(i)));
							else if (field.getType().equals("html"))
								dataObjs.add(getHtml(nodes.get(i)));
							else if (field.getType().equals("image"))
								dataObjs.add(getImage(nodes.get(i)));
						}
						dataObj.append(field.getKey(), dataObjs);
					} else {
						if (field.getType().equals("text"))
							dataObj.append(field.getKey(), getText(nodes.get(0)));
						else if (field.getType().equals("html"))
							dataObj.append(field.getKey(), getHtml(nodes.get(0)));
						else if (field.getType().equals("image"))
							dataObj.append(field.getKey(), getImage(nodes.get(0)));
					}
				}
			}
			return dataObj;
		} catch (Exception e) {
			logger.error(Utils.getTrace(e));
		}
		
		return null;
	}
	
	private String getText(Element node) {
		return node.text();
	}
	
	private String getHtml(Element node) {
		return node.outerHtml();
	}
	
	private BasicDBObject getImage(Element node) {
		BasicDBObject imgObj = new BasicDBObject();
		
		Attributes atrs = node.attributes();
		Map<String, String> atrsMap = atrs.dataset(); 
		for (String key : atrsMap.keySet()) {
			imgObj.put(key, atrsMap.get(key));
		}
		
		return imgObj;
	}
	
	private Object getWithScript(String pageSource, String script, String id) {
		if (jsEngine == null) {
			jsEngine = AppConfiguer.constructJavaScriptEngine();
		}
		String functionName = FlowCrawlExtractorManager.scriptFunctionPrefix+id;
		
		if (!hasExceScriptIdList.contains(id)) {
			hasExceScriptIdList.add(id);
			try {
				String fullScript = "function "+ functionName +"(source) {"+ script +"\n"
						+ " var ret = parse(source);\n"
						+ "var datetype = typeof(ret);\n"
						+ "if ( datetype === 'object' ) { ret = JSON.stringify(ret);}\n"
						+ "	return JSON.stringify({\"type\":datetype, \"result\":ret});}";
				jsEngine.eval(fullScript);
			} catch (ScriptException e) {
				logger.error(Utils.getTrace(e));
				e.printStackTrace();
				hasExceScriptIdList.remove(id);
				return new BasicDBObject();
			}
		}
		
		Invocable jsInvoke = (Invocable) jsEngine;
		
		Object result;
		try {
			result = jsInvoke.invokeFunction(functionName, new Object[] {pageSource});
		} catch (Exception e) {
			logger.error(Utils.getTrace(e));
			return new BasicDBObject();
		}
		
		if (result instanceof java.lang.Double) {
			Double ret = (Double) result;
			return ret;
		} else if (result instanceof java.lang.Boolean) {
			Boolean ret = (Boolean) result;
			return ret;
		} else if (result instanceof java.lang.String) {
			String ret = (String) result;
			JSONObject obj = new JSONObject(ret);
			if (obj.getString("type").toLowerCase().equals("string")) {
				return obj.getString("result");
			} else if (obj.getString("type").toLowerCase().equals("number")) {
				return obj.getDouble("result");
			} else if (obj.getString("type").toLowerCase().equals("object")) {
				return (BasicDBObject) JSON.parse(obj.getString("result"));
			} else {
				logger.info("Script get return date js type : " + obj.getString("type") + " that we can't parse!");
			}
		} else {
			logger.info("Script get return date type : " + result.getClass() + " that we can't parse!");
		}
		
		return new BasicDBObject();
	}
	
	public static void main(String[] argv) throws IOException, InterruptedException, ScriptException, NoSuchMethodException {
		
		ScriptEngine jsEngine = AppConfiguer.constructJavaScriptEngine();
		
		jsEngine.eval("function parse_1(source) { function parse(source) {var o = source; return {'key':12131};}\n"
				+ " var ret = parse(source);\n"
				+ "var datetype = typeof(ret);\n"
				+ "if ( datetype === 'object' ) { ret = JSON.stringify(ret);}\n"
				+ "	return JSON.stringify({\"type\":datetype, \"result\":ret});}"
				);
		Invocable jsInvoke = (Invocable) jsEngine;
		Object result = jsInvoke.invokeFunction("parse_1", new Object[] {"<html></html>"});
		
		if (result.getClass() == java.lang.Double.class) {
			Double ret = (Double) result;
			System.out.println(ret);
		} else if (result.getClass() == java.lang.String.class) {
			String ret = (String) result;
			JSONObject obj = new JSONObject(ret);
			if (obj.getString("type").toLowerCase().equals("string")) {
				
			} else if (obj.getString("type").toLowerCase().equals("object")) {
				
			} else {
				logger.info("Script get return date js type : " + obj.getString("type") + " that we can't parse!");
			}
			System.out.println(ret);
		} else if (result.getClass() == java.lang.Boolean.class) {
			Boolean ret = (Boolean) result;
			System.out.println(ret);
		} else {
			logger.info("Script get return date type : " + result.getClass() + " that we can't parse!");
		}
	}
}
